package com.etlmaster.executor.steps.table;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.processors.Table.CreateMPPTableMeta;
import com.etlmaster.executor.processors.Table.CreateMPPTableProcessor;
import com.etlmaster.executor.processors.Table.CreateTableProcessor;
import com.etlmaster.executor.processors.Table.RationalDbProcessor;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.StepState;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("crtMppTab")
public class CreateMPPTableStep extends Step {
	private CreateMPPTableStepMeta meta;
	private String dsType;

	public CreateMPPTableStep(StepMeta meta) {
		super(meta);
		this.meta = (CreateMPPTableStepMeta) meta;
	}

	@Override
	protected int handle() {
		try {
			dsType = DataCollectionUtils.getDsType(meta.getDsName());
		} catch (Exception e) {
			meta.log("ERROR", "获取数据库类型失败");
			return StepState.RUNNING_FAILED;
		}

		Map<String, Object> data = new HashMap<String, Object>();
		data.putAll(meta.getVariables());
		if (meta.getExtraParam() != null) {
			data.putAll(meta.getExtraParam());
		}
		CreateMPPTableMeta tablemeta = new CreateMPPTableMeta(meta.getTableName(), meta.getXmlId(), dsType,
				meta.getVariable("taskid").toString(), meta.getOverride(), meta.getOptions(), meta.getErrLog(),
				meta.getLog(), meta.getReject(), meta.getServerName());

		tablemeta.setMetaTableName(meta.getMetaTableName());
		tablemeta.setXmlId(meta.getXmlId());
		String result = "";
		CreateTableProcessor processor;
		if (isDdlFromFile()) {
			processor = new CreateMPPTableProcessor();
			result = processor.process(tablemeta, data);
		} else {
			processor = new RationalDbProcessor();
			result = processor.process(tablemeta, data);
		}
		meta.getStepState().getReturnValue().put("ddl", processor.getDdl());
		if (result.equals(StepState.RUNNING_SUCCESSED + "")) {
			return StepState.RUNNING_SUCCESSED;
		} else {
			meta.getStepState().getReturnValue().put("errorMessage", result);
			return StepState.RUNNING_FAILED;
		}
	}

	@Override
	public boolean stop() {
		return false;
	}

	public boolean isDdlFromFile() {
		String[] dbArray = DataCollectionUtils.getProperties("tableDDlList").split(",");
		List<String> dblist = Arrays.asList(dbArray);
		return dblist.contains(dsType);
	}
}
